package cn.com.idmy.mq;

import cn.com.idmy.mq.annotation.MqConsumer;
import cn.com.idmy.mq.annotation.MqConsumerExt;
import cn.com.idmy.mq.annotation.MqSubscribe;
import cn.com.idmy.mq.coder.MqMsgDecoder;
import cn.com.idmy.mq.listener.ConcurrentlyMsgListener;
import cn.com.idmy.mq.listener.OrderlyMsgListener;
import cn.com.idmy.mq.model.enums.MqConsumeMode;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.aop.scope.ScopedProxyUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.core.env.Environment;

import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.*;
import java.util.Map.Entry;
import java.util.stream.Collectors;

@Slf4j
@RequiredArgsConstructor
class MqConsumerManager implements ApplicationContextAware, SmartInitializingSingleton {
    private final MqProps props;
    private final MqMsgDecoder decoder;
    private final Map<String, Map<String, Map<String, Method>>> methodMap = new HashMap<>();
    private final Map<String, Map<String, Map<String, Class<?>>>> classMap = new HashMap<>();
    private final Map<String, Map<String, Map<String, Object>>> beanMap = new HashMap<>();
    private final Map<String, Map<String, Map<String, MqMsgDecoder>>> decoderMap = new HashMap<>();
    private final Map<String, MqConsumer> consumerMap = new HashMap<>();
    private final Map<String, MqConsumerExt> consumerExtMap = new HashMap<>();

    private ConfigurableApplicationContext applicationContext;
    private Environment environment;

    @Override
    public void setApplicationContext(ApplicationContext in) throws BeansException {
        applicationContext = (ConfigurableApplicationContext) in;
        environment = in.getEnvironment();
    }

    @Override
    public void afterSingletonsInstantiated() {
        log.info("mq消费者初始化开始……");
        Map<String, Object> beans = applicationContext.getBeansWithAnnotation(MqConsumer.class)
                .entrySet()
                .stream()
                .filter(entry -> !ScopedProxyUtils.isScopedTarget(entry.getKey()))
                .collect(Collectors.toMap(Entry::getKey, Entry::getValue));
        beans.forEach((key, value) -> {
            Object bean = AopTargetUtil.getTarget(value);
            if (bean != null) {
                Method[] methods = bean.getClass().getDeclaredMethods();
                if (methods.length != 0) {
                    buildMap(methods, bean);
                }
            }
        });
        build();
        log.info("mq消费者初始化结束。");
    }

    private void buildMap(Method[] methods, Object bean) {
        MqConsumer consumer = bean.getClass().getAnnotation(MqConsumer.class);
        for (Method method : methods) {
            MqSubscribe subscribe = method.getAnnotation(MqSubscribe.class);
            if (subscribe != null) {
                String topic = getTopic(subscribe);
                String group = getGroup(consumer);

                methodMap.putIfAbsent(group, new HashMap<>());
                methodMap.get(group).putIfAbsent(topic, new HashMap<>());
                classMap.putIfAbsent(group, new HashMap<>());
                classMap.get(group).putIfAbsent(topic, new HashMap<>());

                Type type = method.getGenericParameterTypes()[0];
                Class<?> parameterClass;
                if (type instanceof Class) {
                    parameterClass = (Class<?>) type;
                } else if (type instanceof ParameterizedType) {
                    parameterClass = (Class<?>) ((ParameterizedType) type).getRawType();
                } else {
                    throw new RuntimeException("parameter type is unknown");
                }

                beanMap.putIfAbsent(group, new HashMap<>());
                beanMap.get(group).putIfAbsent(topic, new HashMap<>());
                decoderMap.putIfAbsent(group, new HashMap<>());
                decoderMap.get(group).putIfAbsent(topic, new HashMap<>());
                consumerMap.putIfAbsent(group, consumer);

                MqConsumerExt ext = bean.getClass().getAnnotation(MqConsumerExt.class);
                if (ext != null) {
                    consumerExtMap.put(group, ext);
                }

                MqMsgDecoder decoder = getMessageDecoder(subscribe);
                Set<String> tagSet = getTags(subscribe);
                tagSet.forEach(tag -> {
                    putMethod(group, topic, tag, method);
                    putClass(group, topic, tag, parameterClass);
                    putBean(group, topic, tag, bean);
                    putDecoder(group, topic, tag, decoder);
                });

                log.info("mq消费者：[tags: {}, topic: {}, consumerGroup: {}, method: {}.{}]", tagSet, topic, group, bean.getClass().getName(), method.getName());
            }
        }

    }

    private void build() {
        buildRocketmqConsumer();
    }

    private void buildRocketmqConsumer() {
        if (CollUtil.isNotEmpty(consumerMap)) {
            Iterator<String> iterator = consumerMap.keySet().iterator();
            while (true) {
                String group;
                do {
                    if (!iterator.hasNext()) {
                        return;
                    }
                    group = iterator.next();
                } while (!consumerMap.containsKey(group));

                MqConsumer annotation = consumerMap.get(group);
                for (int i = 1; i <= getInstanceNumber(annotation); ++i) {
                    IMqConsumer consumer;
                    if (consumerExtMap.containsKey(group)) {
                        consumer = new RocketmqConsumer(props, group, annotation.msgMode(), getConsumerId(i), consumerExtMap.get(group), environment);
                    } else {
                        consumer = new RocketmqConsumer(props, group, annotation.msgMode(), getConsumerId(i));
                    }

                    if (MqConsumeMode.CONCURRENTLY == annotation.consumeMode()) {
                        consumer.setMessageListener(new ConcurrentlyMsgListener(classMap.get(group), methodMap.get(group), beanMap.get(group), decoderMap.get(group)));
                    } else {
                        consumer.setMessageListener(new OrderlyMsgListener(classMap.get(group), methodMap.get(group), beanMap.get(group), decoderMap.get(group)));
                    }

                    String finalGroup = group;
                    beanMap.get(group).keySet().forEach(topic -> consumer.subscribe(topic, getTags(finalGroup, topic)));
                    consumer.start();
                }
            }
        }
    }

    private String getGroup(MqConsumer in) {
        String group = in.group();
        if (group.startsWith("${") && group.endsWith("}")) {
            return environment.resolveRequiredPlaceholders(group);
        } else {
            return StrUtil.blankToDefault(group, props.getConsumerGroup());
        }
    }

    private String getTopic(MqSubscribe in) {
        String topic = in.topic();
        if (topic.startsWith("${") && topic.endsWith("}")) {
            return environment.resolveRequiredPlaceholders(topic);
        } else {
            return StrUtil.blankToDefault(topic, props.getTopic());
        }
    }

    private Set<String> getTags(MqSubscribe in) {
        return new HashSet<>(Arrays.asList(in.value()));
    }

    private Set<String> getTags(String group, String topic) {
        return beanMap.get(group).get(topic).keySet().stream().filter(tag -> !"*".equals(tag)).collect(Collectors.toSet());
    }

    private Integer getInstanceNumber(MqConsumer in) {
        String num = in.instanceNumber();
        return num.startsWith("${") && num.endsWith("}") ? Integer.valueOf(environment.resolveRequiredPlaceholders(num)) : Integer.valueOf(num);
    }

    private MqMsgDecoder getMessageDecoder(MqSubscribe in) {
        Class<? extends MqMsgDecoder>[] clazz = in.decoder();
        return clazz.length > 0 ? applicationContext.getBean(clazz[0]) : decoder;
    }

    private void putMethod(String consumerGroup, String topic, String tag, Method value) {
        if (methodMap.get(consumerGroup).get(topic).containsKey(tag)) {
            throw new RuntimeException("consumerGroup:" + consumerGroup + ", topic:" + topic + ", tag:" + tag + ", associate method already exists");
        } else if (methodMap.get(consumerGroup).get(topic).containsKey("*")) {
            throw new RuntimeException("consumerGroup:" + consumerGroup + ", topic:" + topic + ", tag:" + tag + ", associate method already exists");
        } else if (methodMap.get(consumerGroup).get(topic).isEmpty()) {
            methodMap.get(consumerGroup).get(topic).put(tag, value);
        } else if (!"*".equals(tag)) {
            methodMap.get(consumerGroup).get(topic).put(tag, value);
        } else {
            throw new RuntimeException("consumerGroup:" + consumerGroup + ", topic:" + topic + ", tag:" + tag + ", associate method already exists");
        }
    }

    private void putClass(String consumerGroup, String topic, String tag, Class<?> value) {
        if (classMap.get(consumerGroup).get(topic).containsKey(tag)) {
            throw new RuntimeException("consumerGroup:" + consumerGroup + ", topic:" + topic + ", tag:" + tag + ", associate class already exists");
        } else if (classMap.get(consumerGroup).get(topic).containsKey("*")) {
            throw new RuntimeException("consumerGroup:" + consumerGroup + ", topic:" + topic + ", tag:" + tag + ", associate class already exists");
        } else if (classMap.get(consumerGroup).get(topic).isEmpty()) {
            classMap.get(consumerGroup).get(topic).put(tag, value);
        } else if (!"*".equals(tag)) {
            classMap.get(consumerGroup).get(topic).put(tag, value);
        } else {
            throw new RuntimeException("consumerGroup:" + consumerGroup + ", topic:" + topic + ", tag:" + tag + ", associate class already exists");
        }
    }

    private void putBean(String group, String topic, String tag, Object value) {
        if (beanMap.get(group).get(topic).containsKey(tag)) {
            throw new RuntimeException("consumerGroup:" + group + ", topic:" + topic + ", tag:" + tag + ", associate bean already exists");
        } else if (beanMap.get(group).get(topic).containsKey("*")) {
            throw new RuntimeException("consumerGroup:" + group + ", topic:" + topic + ", tag:" + tag + ", associate bean already exists");
        } else if (beanMap.get(group).get(topic).isEmpty()) {
            beanMap.get(group).get(topic).put(tag, value);
        } else if (!"*".equals(tag)) {
            beanMap.get(group).get(topic).put(tag, value);
        } else {
            throw new RuntimeException("consumerGroup:" + group + ", topic:" + topic + ", tag:" + tag + ", associate bean already exists");
        }
    }

    private void putDecoder(String group, String topic, String tag, MqMsgDecoder value) {
        if (decoderMap.get(group).get(topic).containsKey(tag)) {
            throw new RuntimeException("consumerGroup:" + group + ", topic:" + topic + ", tag:" + tag + ", associate decoder already exists");
        } else if ((decoderMap.get(group)).get(topic).containsKey("*")) {
            throw new RuntimeException("consumerGroup:" + group + ", topic:" + topic + ", tag:" + tag + ", associate decoder already exists");
        } else if (!decoderMap.get(group).get(topic).isEmpty()) {
            if ("*".equals(tag)) {
                throw new RuntimeException("consumerGroup:" + group + ", topic:" + topic + ", tag:" + tag + ", associate decoder already exists");
            } else {
                decoderMap.get(group).get(topic).put(tag, value);
            }
        } else {
            decoderMap.get(group).get(topic).put(tag, value);
        }
    }

    private static String getConsumerId(int i) {
        return "consumer-" + i + "_" + IdUtil.fastSimpleUUID();
    }
}
